package com.example.mq2.mqserver.datacenter;

import com.example.mq2.common.Mq2Exception;
import com.example.mq2.mqserver.core.Binding;
import com.example.mq2.mqserver.core.Exchange;
import com.example.mq2.mqserver.core.MSGQueue;
import com.example.mq2.mqserver.core.Message;
import org.springframework.context.MessageSource;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
* 使用这个类来同一管理内存中的数据*/
public class MemoryDataCenter {
    // 交换机
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // 队列
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 绑定
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 消息
    // MessageID和 message 对象的关联关系
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // message 对象和队列之间的关系
    // 第一个String 是queueName
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap();
    // 第一个String 是队列的名字，第二个String 是MessageID，此处的设计是为了后续手动应答的时候要删除队列中对应的消息
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // 针对交换机的增删改查
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
    }
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
    }

    // 针对队列的增删改查
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
    }
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
    }

    // 针对交换机的增删改查
    public void insertBinding(Binding binding) throws Mq2Exception {
        // 首先 exchangeName 查一下，对应的hashMap是否存在，如果不存在就创建一个
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        // 这个方法就相当于上述的逻辑
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        synchronized (bindingMap) { // 锁对象为binding，此时针对哪个bindingMap来操作就针对哪一个进行加锁即可
            // 再根据 queueName 查一下，如果这个binding 不存在才能进行插入
            if (bindingMap.get(binding.getQueueName()) != null) {
                // 此时说明binding已经存在，就抛出一个异常
                throw new Mq2Exception("[MemoryDataCenter] 绑定已存在在！exchangeName= " + binding.getExchangeName()
                        + ", queueName= " + binding.getQueueName());
            }
            // 如果不存在再插入binding
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 绑定添加成功！exchangeName= " + binding.getExchangeName() +
                    "queueName= " + binding.getQueueName());
        }
        // 上述代码是两个操作，单独拿出来一个操作确实是线程安全的，但是此时上述两个操作是有印尼过关系的，如果有两个线程来插入，线程1判定了
        // 当前binding 是不存在的，此时没等线程1插入呢，线程2也来插入，也判定当前binding不存在，线程2插入了，此时就是线程不安全的了
    }

    // 获取绑定：写两个版本：
    // 1. 根据 exchangeName 和 queueName 查询出唯一的一个 binding
    // 2. 根据 exchangeName查询出所有的 binding 对象
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) return null;
        return bindingMap.get(queueName);
    }
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 删除binding
    public void deleteBinding(Binding binding) throws Mq2Exception {
        // 先根据 ExchangeName 查询一下当前的 binding 是否存在
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 要删除的binding 不存在  报错就行
            throw new Mq2Exception("[MemoryDataCenter] 当前要删除的绑定不存在！ exchangeName= " + binding.getExchangeName()
                + ", queueName= " + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功！exchangeName= " + binding.getExchangeName() +
        ", queueName= " + binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加消息成功！ messageId= " + message.getMessageId());
    }

    // 根据id获取一条消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据id删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除！messageId= " + messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 把消息放到对应的队列数据结构中
        // 先根据队列的名字找到对应的链表
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k-> new LinkedList<>());
        // 上述代码的逻辑就是现根据queueName查到对应的Messages，如果这个链表不存在，就创建一个新的链表，最后进行返回
        synchronized (messages) {
            messages.add(message);
        }
        // 在这里也把message插入到消息中心中一份，因为此时也不确定这个消息是否已经在messageMap中存在
        // 即使存在了也没有关系, 插入的消息会覆盖掉原来的消息，但是消息的内容是一样的
        addMessage(message);
    }

    // 从队列中取消息
    public Message pollMessage(String queueName) {
        // 根据 queueName 先查询一下对应的链表是否存在
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) return null; // 此时还需要把这个条件单拿出来，不能对空的messages进行加锁
        // 链表中有元素，就进行头删
        synchronized (messages) {
            Message curMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！ messageId= " + curMessage.getMessageId());
            return curMessage;
        }
    }

    // 获取指定队列中的消息个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) return 0;
        synchronized (messages) {
            return messages.size();
        }
    }

    // 针对未确认消息进行操作
    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        // 先查一下queue对应的hashmap是否存在
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k-> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加未确认消息成功！ messageId= " + message.getMessageId());
    }

    // 删除未确认的消息 (消息已经被确认了)
    public void deleteMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) return;
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 删除未确认消息成功！ messageId= " + messageId);
    }

    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) return null;
        return messageHashMap.get(messageId);
    }

    // 这个方法就是读取硬盘上所有的数据，从硬盘上之前持久化的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws Mq2Exception, IOException, ClassNotFoundException {
        // 1. 清空所有内存中的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        queueMessageMap.clear();
        messageMap.clear();
        queueMessageWaitAckMap.clear();
        // 2. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 3. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 4. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for (Binding binding : bindings) {
            // 先根据交换机的名字来查询下binding对应的map是否存在，如果不存在就创建一个新的 ConcurrentHashMap
            // 有了内部的那个map之后再把binding插入到这个内部的map中即可
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 5. 恢复所有的消息数据
        // 遍历所有的队列，根据队列的名字来获取所有的消息数据
        for (MSGQueue queue : queues) {
            // 注意这个方法读到文件的末尾就会抛出一个异常，此时就证明消息读完了，然后返回这个链表
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            // 还需要把消息数据添加到消息中心里边
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
    }
}
